package com.niit.spark.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Date:2025/5/22
 * Author：Ys
 * Description:
 */
object StreamingTransform04 {

  def main(args: Array[String]): Unit = {
    val ssc = new StreamingContext(new SparkConf().setMaster("local[*]").setAppName("StreamingTransform04"), Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    /*
      Spark Streaming 在有些地方功能不完善 或者是我们不习惯使用DStream，那么Spark Streaming提供了transform方法，
       可以将DStream里面的数据转换成 RDD，转成可以RDD后就可以用行动算子、转换算子。最后返回数据还是DStream。
     */

    //以为单词统计为例
    val wordDS: DStream[(String, Int)] = lines.transform(rdd => {
      val words: RDD[String] = rdd.flatMap(_.split(" "))
      val wordOne: RDD[(String, Int)] = words.map((_, 1))
      val resWord: RDD[(String, Int)] = wordOne.reduceByKey(_ + _)
      resWord
    })
    //Spark Streaming 必须要写输出，否则不会执行
    wordDS.print()

    ssc.start()
    ssc.awaitTermination()

  }

}
